package org.leiyang.loader;

import org.leiyang.entries.BlockFileMetaInfo;
import org.leiyang.entries.TempFileIterator;
import org.leiyang.entries.Record;

import java.io.*;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class SourceDataPreLoad {
    private static final Logger log = LoggerFactory.getLogger(SourceDataPreLoad.class);

    private static final int BLOCK_SIZE = 1000000;  //每个小文件的数据条数
    //源数据文件(约10亿条数据)
//    private static final String SOURCE_INPUT_FILE = "/root/source_data/source_data.txt";
    private static final String SOURCE_INPUT_FILE = "E:\\source_data\\data.txt";

//    public static final String BLOCK_FILE = "/root/source_data/blocks/";
    public static final String BLOCK_FILE = "E:\\source_data\\blocks\\";
    //临时文件存放路径
//    private static final String TEMP_DIR = "/root/source_data/temp/";
    private static final String TEMP_DIR = "E:\\source_data\\temp\\";
    //小文件名前缀
    private static final String BLOCK_FILE_PREFIX = "s_block_";
    //存储所有的小文件元信息
    private static final List<BlockFileMetaInfo> block_files_meta_list = new ArrayList<>(1000);
    //用于标记源数据是否可用
    private static boolean lockSource = false;

    public SourceDataPreLoad() throws IOException {
        Files.createDirectories(Paths.get(BLOCK_FILE));
        Files.createDirectories(Paths.get(TEMP_DIR));
    }

    public static List<BlockFileMetaInfo> blockFilesMetaInfoGot(){
        return block_files_meta_list;
    }

    /**
     * 重置源数据
     * 前提条件: SOURCE_INPUT_FILE下的源数据文件已更新成最新版
     */
    public static void reSetSourceData() {
        //将BLOCK_FILE目录下的文件删除
        lockSource = true;
        Path dir = Paths.get(BLOCK_FILE);
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(dir, BLOCK_FILE_PREFIX.concat("*.dat"))) {
            for (Path entry : stream) {
                if (Files.isRegularFile(entry)) {
                    Files.delete(entry);
                }
            }
            block_files_meta_list.clear();
            preLoadSourceData();
        } catch (IOException E){
            log.error("重置源数据失败, ", E);
        } finally {
            lockSource = false;
        }
    }

    /**
     * 源数据是否可用
     */
    public static boolean blockFileAvailable() {
        return !lockSource;
    }

    /**
     * 预加载源数据
     */
    public static void preLoadSourceData() throws IOException {
        Path dir = Paths.get(BLOCK_FILE);
        // 过滤出BLOCK_FILE_PREFIX前缀的文件
        try (DirectoryStream<Path> stream = Files.newDirectoryStream(dir, BLOCK_FILE_PREFIX.concat("*.dat"))) {
            for (Path entry : stream) {
                if (Files.isRegularFile(entry)) {
                    String bFileName = entry.getFileName().toString();
                    String[] subNames = bFileName.split("_");
                    Integer maxId = Integer.parseInt(subNames[3].replaceAll(".dat", ""));
                    BlockFileMetaInfo metaInfo = BlockFileMetaInfo.builder().minId(Integer.parseInt(subNames[2]))
                            .maxId(maxId)
                            .fileName(bFileName)
                            .build();
                    block_files_meta_list.add(metaInfo);
                }
            }
        }
        if(block_files_meta_list.isEmpty()) {
            try {
                //需要重新切分源文件
                List<Path> tempFiles = splitSourceDataToTempFiles();
                mergeAndCreateBlockFiles(tempFiles);
                cleanTempFiles(tempFiles);
            } catch (IOException e) {
                log.error("切分源数据失败，请确认源数据文件是否存在.", e);
            }
        }
    }

    /**
     * 将源文件切分成临时小文件
     */
    private static List<Path> splitSourceDataToTempFiles() throws IOException {
        List<Path> tempFiles = new ArrayList<>(1000);
        List<Record> buffer = new ArrayList<>(BLOCK_SIZE);
        try (BufferedReader reader = Files.newBufferedReader(Paths.get(SOURCE_INPUT_FILE))) {
            String line;
            int fileCounter = 0;
            int index = 0;

            while ((line = reader.readLine()) != null) {
                String[] parts = line.split(",");
                buffer.add(new Record(Integer.parseInt(parts[0]), Integer.parseInt(parts[1]), Float.parseFloat(parts[2])));
                index++;
                if (index >= BLOCK_SIZE) {
                    tempFiles.add(sortAndSaveTempFile(buffer, fileCounter++));
                    index = 0;
                    buffer.clear();
                }
            }

            if (!buffer.isEmpty()) {
                tempFiles.add(sortAndSaveTempFile(buffer, fileCounter));
            }
        }
        return tempFiles;
    }

    /**
     *  读取临时文件并生成最终的BLOCK小文件
     * @param tempFiles 临时小文件
     */
    private static void mergeAndCreateBlockFiles(List<Path> tempFiles) throws IOException {
        int tempFileCount = tempFiles.size();
        //存储临时文件的遍历器
        PriorityQueue<TempFileIterator> pq = new PriorityQueue<>(tempFileCount);
        //用于最后关闭流
        List<ObjectInputStream> streams = new ArrayList<>(tempFileCount);
        for(Path file : tempFiles) {
            try {
                ObjectInputStream ois = new ObjectInputStream(
                        new BufferedInputStream(Files.newInputStream(file)));
                streams.add(ois);
                pq.add(new TempFileIterator(ois));
            } catch (Exception e){
                log.error("读取临时文件:{}异常，跳过这个临时文件.", file.getFileName(), e);
            }
        }

        //使用多路归并进行排序
        List<Record> blockBuffer = new ArrayList<>(BLOCK_SIZE);
        int blockCounter = 0;
        int minIdInBlock = Integer.MAX_VALUE;
        int maxIdInBlock = Integer.MIN_VALUE;

        while (!pq.isEmpty()) {
            TempFileIterator iterator = pq.poll();
            Record record = iterator.getCurrent();

            // 更新当前块的ID范围
            if (record.getId() < minIdInBlock) minIdInBlock = record.getId();
            if (record.getId() > maxIdInBlock) maxIdInBlock = record.getId();

            blockBuffer.add(record);

            // 读取下一条记录
            if (iterator.readNext()) {
                pq.add(iterator);
            }

            // 输出完整数据块
            if (blockBuffer.size() >= BLOCK_SIZE) {
                String fName = writeBlock(blockBuffer, blockCounter++, minIdInBlock, maxIdInBlock);
                BlockFileMetaInfo metaInfo = BlockFileMetaInfo.builder().minId(minIdInBlock)
                        .maxId(maxIdInBlock)
                        .fileName(fName)
                        .build();
                //缓存该文件的元信息
                block_files_meta_list.add(metaInfo);

                blockBuffer.clear();
                minIdInBlock = Integer.MAX_VALUE;
                maxIdInBlock = Integer.MIN_VALUE;
            }
        }

        // 输出剩余数据
        if (!blockBuffer.isEmpty()) {
            String fName = writeBlock(blockBuffer, blockCounter, minIdInBlock, maxIdInBlock);
            BlockFileMetaInfo metaInfo = BlockFileMetaInfo.builder().minId(minIdInBlock)
                    .maxId(maxIdInBlock)
                    .fileName(fName)
                    .build();
            //缓存该文件的元信息
            block_files_meta_list.add(metaInfo);
        }

        // 关闭所有流
        for (ObjectInputStream ois : streams) {
            ois.close();
        }
    }

    /**
     * 生成BLOCK二进制小文件，命名规则: s_block_{起始id}_{截止id}
     * @return 小文件名
     */
    private static String writeBlock(List<Record> block, int blockIndex,
                                   int minId, int maxId) throws IOException {
        String filename = String.format(BLOCK_FILE_PREFIX.concat("%d_%d.dat"), minId, maxId);
        Path outputPath = Paths.get(BLOCK_FILE, filename);

        try (ObjectOutputStream oos = new ObjectOutputStream(
                new BufferedOutputStream(Files.newOutputStream(outputPath)))) {
            for (Record r : block) {
                oos.writeInt(r.getId());
                oos.writeInt(r.getInventory());
                oos.writeFloat(r.getPrice());
            }
            log.info("已成功生成第{}个小文件.", blockIndex);
        }
        return filename;
    }

    /**
     * 清除生成的临时小文件
     */
    private static void cleanTempFiles(List<Path> tempFiles) throws IOException {
        for (Path file : tempFiles) {
            Files.deleteIfExists(file);
        }
        log.info("临时小文件已全部删除.");
    }


    /**
     * 排序并生成临时文件
     */
    private static Path sortAndSaveTempFile(List<Record> records, Integer tempFileIndex) throws IOException {
        //先按照id进行排序
        Collections.sort(records);
        Path tempFile = Paths.get(TEMP_DIR, "temp_" + tempFileIndex + ".dat");

        try (ObjectOutputStream oos = new ObjectOutputStream(
                new BufferedOutputStream(Files.newOutputStream(tempFile)))) {
            for (Record r : records) {
                oos.writeInt(r.getId());
                oos.writeInt(r.getInventory());
                oos.writeFloat(r.getPrice());
            }
        }
        return tempFile;
    }
}
